package cn.spark.streaming;


import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.*;

import org.apache.spark.streaming.kafka.KafkaUtils;
import scala.Tuple2;

/**
 * Created by Niko Feng on 4/29/2017.
 */

/**
 * Spark Streaming for Twitter Source Region Computation
**/
public class WindowHotWord {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                .setMaster("yarn-cluster")
                .setAppName("WindowHotWord");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));

        Map<String, Integer> topicThreadMap = new HashMap<String, Integer>();
        topicThreadMap.put(args[0].toString(), 1);

        JavaPairReceiverInputDStream<String, String> lines = KafkaUtils.createStream(
                jssc,
                "student39-x1:2181,student39-x2:2181,student40-x1:2181,student40-x2:2181",
                "DefaultConsumerGroup",
                topicThreadMap
        );

        JavaDStream<String> searchWordsDStream = lines.map(new Function<Tuple2<String, String>, String>() {
            @Override
            public String call(Tuple2<String, String> tuple) throws Exception {
                return tuple._2().split("\\|\\|")[2];
            }
        });


        //Transform the word to the tuple format (word, 1)
        JavaPairDStream<String, Integer> searchWordPairDStream = searchWordsDStream.mapToPair(

                new PairFunction<String, String, Integer>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<String, Integer> call(String searchWord)
                            throws Exception {
                        return new Tuple2<String, Integer>(searchWord, 1);
                    }

                });

        //the program computes every 10 seconds to collect the last 60 seconds RDD data, 
        //which would let us to control the size of DStream and avoid the intensive 
        //computation during the streaming process.
        JavaPairDStream<String, Integer> searchWordCountsDStream =

                searchWordPairDStream.reduceByKeyAndWindow(new Function2<Integer, Integer, Integer>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Integer call(Integer v1, Integer v2) throws Exception {
                        return v1 + v2;
                    }

                }, Durations.seconds(60), Durations.seconds(10));


        //Sort the word and get the Top 3 Language in Twitter
        JavaPairDStream<String, Integer> finalDStream = searchWordCountsDStream.transformToPair(

                new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public JavaPairRDD<String, Integer> call(
                            JavaPairRDD<String, Integer> searchWordCountsRDD) throws Exception {

                        // First we need to invert the word and the count
                        JavaPairRDD<Integer, String> countSearchWordsRDD = searchWordCountsRDD
                                .mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {

                                    private static final long serialVersionUID = 1L;

                                    @Override
                                    public Tuple2<Integer, String> call(
                                            Tuple2<String, Integer> tuple)
                                            throws Exception {
                                        return new Tuple2<Integer, String>(tuple._2, tuple._1);
                                    }
                                });

                        // Sort the word with descending order
                        JavaPairRDD<Integer, String> sortedCountSearchWordsRDD = countSearchWordsRDD
                                .sortByKey(false);

                        // Invert again to transform the tuple as (word, sumcount)
                        JavaPairRDD<String, Integer> sortedSearchWordCountsRDD = sortedCountSearchWordsRDD
                                .mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {

                                    private static final long serialVersionUID = 1L;

                                    @Override
                                    public Tuple2<String, Integer> call(
                                            Tuple2<Integer, String> tuple)
                                            throws Exception {
                                        return new Tuple2<String, Integer>(tuple._2, tuple._1);
                                    }

                                });

                        // Get the Top 3 Language Used for the certain Keyword in Twitter
                        List<Tuple2<String, Integer>> hogSearchWordCounts =
                                sortedSearchWordCountsRDD.take(3);
                        String sql = "INSERT INTO wordcount(word,count) "
                                + "VALUES(?, ?) "
                                + "ON DUPLICATE KEY UPDATE word=?, count=?;";
                        Connection conn = ConnectionPool.getConnection();
                        for (Tuple2<String, Integer> wordCount : hogSearchWordCounts) {
                            PreparedStatement prestmt = conn.prepareStatement(sql);
                            prestmt.setString(1, wordCount._1);
                            prestmt.setInt(2, wordCount._2);
                            prestmt.setString(3, wordCount._1);
                            prestmt.setInt(4, wordCount._2);
                            prestmt.executeUpdate();
                            System.out.println("Successfully inserted one record...");
                            System.out.println(wordCount._1 + ": " + wordCount._2);
                        }

                        return searchWordCountsRDD;
                    }

                });

        finalDStream.print();
        try {
            jssc.start();
            jssc.awaitTermination();
            jssc.close();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            jssc.close();
        }

    }

}

